package com.feidee.fdhadoop.hbase;

import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdhadoop.configUpdater.ConfigUpdater;

import java.io.IOException;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HBaseUtilHelper {
    protected static final Logger logger = LoggerFactory.getLogger(HBaseUtilHelper.class);

    protected static String DEFAULT_FAMILY = "f";
    protected static String DEFAULT_QUALIFIER = "q";

    protected static String get(String rowKey, String family, String qualifier, Table t)
        throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
        Result r = t.get(get);
        if (r.isEmpty() || r.listCells().size() == 0) {
            return null;
        }
        return Bytes.toString(CellUtil.cloneValue(r.listCells().get(0)));
    }

    protected static Map<String, Map<String, String>> get(String rowKey, Table table) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        Result r = table.get(get);
        if (r.isEmpty() || r.listCells().size() == 0) {
            return null;
        }
        Map<String, Map<String, String>> results = new HashMap<>();
        for (Cell cell : r.listCells()) {
            String familyName = Bytes.toString(CellUtil.cloneFamily(cell));
            results.computeIfAbsent(familyName, k -> new HashMap<>())
                .put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
        }
        return results;
    }

    protected static Map<String, String> get(String rowKey, String family, Table table)
        throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        get.addFamily(Bytes.toBytes(family));
        Result r = table.get(get);
        if (r.isEmpty() || r.listCells().size() == 0) {
            return null;
        }
        Map<String, String> result = new HashMap<>();
        for (Cell cell : r.listCells()) {
            result.put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
        }
        return result;
    }

    protected static Map<String, String> get(String rowKey, String family, Set<String> qualifierSet,
        Table t) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        for (String columnName : qualifierSet) {
            get.addColumn(Bytes.toBytes(family), Bytes.toBytes(columnName));
        }
        Result r = t.get(get);
        if (r.isEmpty() || r.listCells().size() == 0) {
            return null;
        }
        Map<String, String> result = new HashMap<>();
        for (Cell cell : r.listCells()) {
            result.put(Bytes.toString(CellUtil.cloneQualifier(cell)), Bytes.toString(CellUtil.cloneValue(cell)));
        }
        return result;
    }

    protected static void put(String rowkey, String family, Map<String, String> valueMap, Table t)
        throws IOException {
        Put put = new Put(Bytes.toBytes(rowkey));
        Iterator<String> iterator = valueMap.keySet().iterator();
        while (iterator.hasNext()) {
            String key = iterator.next();
            String value = valueMap.get(key);
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(key), Bytes.toBytes(value));
        }
        t.put(put);
    }

    protected static void put(String family, List<Map<String, String>> recordList, Table t)
        throws IOException {
        ArrayList<Put> puts = new ArrayList<>();
        for (Map<String, String> record : recordList) {
            String rowkey = record.get("rowkey");
            if (rowkey != null) {
                Put put = new Put(Bytes.toBytes(rowkey));
                Iterator<String> iterator = record.keySet().iterator();
                while (iterator.hasNext()) {
                    String key = iterator.next();
                    if (!key.equals("rowkey")) {
                        String value = record.get(key);
                        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(key), Bytes.toBytes(value));
                    }
                }
                puts.add(put);
            }
        }
        t.put(puts);
    }

    public static void setDefaultFamily(String family) {
        DEFAULT_FAMILY = family;
    }

    public static void setDefaultQualifier(String qualifier) {
        DEFAULT_QUALIFIER = qualifier;
    }
}
